Amazon SageMaker AutopilotのチュートリアルのJob関連の設定を変更して動かしてみた

Amazon SageMaker Autopilotをチュートリアルを元に動かしてみた！

#Amazon SageMaker

#SageMaker Autopilot

#機械学習

#AWS

クルトン

2023.06.08

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

皆さん、こんにちは。

クルトンです！

今回は、Amazon SageMaker Autopilot(以下、Autopilot)をチュートリアルを見て動かしてみました。Autopilotでは、複数のモデルを学習させ、一番より良い予測精度をもったモデルを使った予測も行なえます。

前提

公式チュートリアルを動かしています。

実行環境は、SageMaker StudioのPython 3 (Data Science)カーネルです。

準備

まずは各メソッドを呼び出すのに必要なboto3など、必要なモジュールをimportします。変数prefixについては、チュートリアルそのままとしていますが、別名を設定しても良いです。ここを変更する事で、S3へデータをアップロードする時のパスが変わります。

import sagemaker
import boto3
from sagemaker import get_execution_role

region = boto3.Session().region_name

session = sagemaker.Session()
bucket = session.default_bucket()
prefix = "sagemaker/autopilot-dm"

role = get_execution_role()

sm = boto3.Session().client(service_name="sagemaker", region_name=region)

分析対象のファイル準備

Autopilotで学習させるデータを用意します。wgetコマンドを使って公式が用意しているデータをダウンロードします。公式ドキュメントによると、CSVファイルかParquetファイルをサポートしているとの事なので、ダウンロードしたzipファイルを解凍して、CSVファイルを取り出す必要があります。

!apt-get install unzip
# 同じ階層に保存(ファイル名: bank-additonal.zip)
!wget -N https://sagemaker-sample-data-us-west-2.s3-us-west-2.amazonaws.com/autopilot/direct_marketing/bank-additional.zip
!unzip -o bank-additional.zip

local_data_path = "./bank-additional/bank-additional-full.csv"

Amazon S3へファイルアップロード

ダウンロードしてきたデータを、学習用とテスト用の2つに分け、Autopilotで扱えるようにS3へデータをアップロードします。

データ読み込み

CSVファイルを読み込むため、pandasモジュールのread_csvメソッドを使います。

import pandas as pd

data = pd.read_csv(local_data_path)
pd.set_option("display.max_columns", 500)  # 全カラムを見えるよう設定
pd.set_option("display.max_rows", 10)  # 1ページで収まるように行数を設定
data # データを出力して確認

データ分割

次に、データを学習用とテスト用の2つに分割します。

train_data = data.sample(frac=0.8, random_state=200)

test_data = data.drop(train_data.index)

test_data_no_target = test_data.drop(columns=["y"])

S3へアップロード

最後に、分割したデータをS3へアップロードします。学習用のデータとテスト用のデータについて、headerの有無の設定が異なるので、他のデータを使う場合に注意が必要そうです。

train_file = "train_data.csv"
train_data.to_csv(train_file, index=False, header=True)
train_data_s3_path = session.upload_data(path=train_file, key_prefix=prefix + "/train")
print("Train data uploaded to: " + train_data_s3_path)

test_file = "test_data.csv"
test_data_no_target.to_csv(test_file, index=False, header=False)
test_data_s3_path = session.upload_data(path=test_file, key_prefix=prefix + "/test")
print("Test data uploaded to: " + test_data_s3_path)

AutopilotのJob設定

ここでは、AutopilotでのJobに関連する設定を行ないます。次の公式ドキュメントが参考になります。

変数auto_ml_job_configで設定しているCompletionCriteriaでは、他にも1つのJobが完了するまでの時間を制限する事も可能です。

auto_ml_job_config = {"CompletionCriteria": {"MaxCandidates": 10}}

input_data_config = [
    {
        "DataSource": {
            "S3DataSource": {
                "S3DataType": "S3Prefix",
                "S3Uri": f"s3://{bucket}/{prefix}/train",
            }
        },
        "TargetAttributeName": "y",
    }
]

output_data_config = {"S3OutputPath": f"s3://{bucket}/{prefix}/output"}

注意点

データの前処理とハイパーパラメータのTrainingJobの2種類が動き、CompletionCriteria で設定したMaxCandidatesの数に2を掛けた20個のTrainingJob数が動きました。また、関連して、ProcessingJobが複数動きます。 「とにかくコストを小さくして動かしたい」といった場合は、3など小さな数字を設定する事が可能です。ただし、予測の精度が落ちる可能性ありますので、ご留意ください。

AutopilotのJobを開始

実際にAutopilotを使って、機械学習モデルの学習とハイパーパラメータのチューニングを行ないます。どのような予測をさせるのかが分かっている場合は、create_auto_ml_jobメソッドのProblemTypeで指定できます。指定できるタイプは、BinaryClassification(2値分類)とMulticlassClassification(多クラス分類)とRegression(回帰)の3つです。

from time import gmtime, strftime, sleep

timestamp_suffix = strftime("%d-%H-%M-%S", gmtime())

auto_ml_job_name = "automl-banking-" + timestamp_suffix
print("AutoMLJobName: " + auto_ml_job_name)

sm.create_auto_ml_job(
    AutoMLJobName=auto_ml_job_name,
    InputDataConfig=input_data_config,
    OutputDataConfig=output_data_config,
    AutoMLJobConfig=auto_ml_job_config,
    RoleArn=role,
)

Jobの詳細を出力

create_auto_ml_jobメソッドを動かしたのち、各Jobの状態を次のコードを使うと確認できます。マネージメントコンソールからも、学習中などのJobのステータスが確認可能です。

print("JobStatus - Secondary Status")
print("------------------------------")


describe_response = sm.describe_auto_ml_job(AutoMLJobName=auto_ml_job_name)
print(describe_response["AutoMLJobStatus"] + " - " + describe_response["AutoMLJobSecondaryStatus"])
job_run_status = describe_response["AutoMLJobStatus"]

while job_run_status not in ("Failed", "Completed", "Stopped"):
    describe_response = sm.describe_auto_ml_job(AutoMLJobName=auto_ml_job_name)
    job_run_status = describe_response["AutoMLJobStatus"]

    print(
        describe_response["AutoMLJobStatus"] + " - " + describe_response["AutoMLJobSecondaryStatus"]
    )
    sleep(30)

Jobのステータスは次のようになりました。出力例としてご確認ください。

JobStatus - Secondary Status
------------------------------
InProgress - AnalyzingData
InProgress - AnalyzingData
InProgress - AnalyzingData
InProgress - AnalyzingData
InProgress - AnalyzingData
……
InProgress - AnalyzingData
InProgress - FeatureEngineering
InProgress - FeatureEngineering
InProgress - FeatureEngineering
……
InProgress - FeatureEngineering
InProgress - ModelTuning
InProgress - ModelTuning
InProgress - ModelTuning
……
InProgress - ModelTuning
InProgress - MergingAutoMLTaskReports
InProgress - MergingAutoMLTaskReports
InProgress - MergingAutoMLTaskReports
……
InProgress - MergingAutoMLTaskReports
Completed - Completed

学習結果確認

複数のモデルを学習した結果から、一番予測精度が高いモデルを確認します。

best_candidate = sm.describe_auto_ml_job(AutoMLJobName=auto_ml_job_name)["BestCandidate"]
best_candidate_name = best_candidate["CandidateName"]
print(best_candidate)
print("\n")
print("CandidateName: " + best_candidate_name)
print(
    "FinalAutoMLJobObjectiveMetricName: "
    + best_candidate["FinalAutoMLJobObjectiveMetric"]["MetricName"]
)
print(
    "FinalAutoMLJobObjectiveMetricValue: "
    + str(best_candidate["FinalAutoMLJobObjectiveMetric"]["Value"])
)

次以降のコードでは、ここで確認したモデルを使って推論を実行します。

バッチ推論を実施

まずは推論をさせるためのモデルを、create_modelメソッドでデプロイします。

model_name = "automl-banking-model-" + timestamp_suffix

model = sm.create_model(
    Containers=best_candidate["InferenceContainers"], ModelName=model_name, ExecutionRoleArn=role
)

print(f"Model ARN corresponding to the best candidate is : {model['ModelArn']}")

バッチ推論を実施するためtransform_jobを設定後実行

次に、バッチ推論を行なうにあたって、Transformジョブを実行するので、推論を実行するファイルや推論結果を保存するS3 URIを設定します。

transform_job_name = "automl-banking-transform-" + timestamp_suffix

transform_input = {
    "DataSource": {"S3DataSource": {"S3DataType": "S3Prefix", "S3Uri": test_data_s3_path}},
    "ContentType": "text/csv",
    "CompressionType": "None",
    "SplitType": "Line",
}

transform_output = {
    "S3OutputPath": f"s3://{bucket}/{prefix}/inference-results",
}

transform_resources = {"InstanceType": "ml.m5.4xlarge", "InstanceCount": 1}

sm.create_transform_job(
    TransformJobName=transform_job_name,
    ModelName=model_name,
    TransformInput=transform_input,
    TransformOutput=transform_output,
    TransformResources=transform_resources,
)

次のコードで、推論の途中経過を確認できます。

print("JobStatus")
print("----------")


describe_response = sm.describe_transform_job(TransformJobName=transform_job_name)
job_run_status = describe_response["TransformJobStatus"]
print(job_run_status)

while job_run_status not in ("Failed", "Completed", "Stopped"):
    describe_response = sm.describe_transform_job(TransformJobName=transform_job_name)
    job_run_status = describe_response["TransformJobStatus"]
    print(job_run_status)
    sleep(30)

推論の途中経過の出力例は、次のようになります。

JobStatus
----------
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
InProgress
Completed

推論結果を確認

バッチ推論が完了すると、出力先に設定したパスへファイルがアップロードされています。ここでは、S3から推論結果のファイルをダウンロードし、pandasモジュールのread_csvメソッドで読み込み後、データの確認をしています。

<br />s3_output_key = f"{prefix}/inference-results/test_data.csv.out"
local_inference_results_path = "inference_results.csv"

s3 = boto3.resource("s3")
inference_results_bucket = s3.Bucket(session.default_bucket())

inference_results_bucket.download_file(s3_output_key, local_inference_results_path)

data = pd.read_csv(local_inference_results_path, sep=";")
pd.set_option("display.max_rows", 10)  # 1ページで収まるように行数を設定
data

リソース削除

Autopilotでは、さまざまなファイルを作成されるので、それらファイルが必要なければ、次のコードで削除できます。

s3 = boto3.resource('s3')
bucket = s3.Bucket(bucket)

job_outputs_prefix = '{}/output/{}'.format(prefix,auto_ml_job_name)
bucket.objects.filter(Prefix=job_outputs_prefix).delete()

その他

全モデルの学習結果を確認する方法や、Autopilotで実行したジョブが終わると2つのノートブックが作成されます。

全モデルの学習結果を確認

次のコードで、一番良かった学習結果のモデル以外では、どのような学習結果だったかを確認できます。

candidates = sm.list_candidates_for_auto_ml_job(
    AutoMLJobName=auto_ml_job_name, SortBy="FinalObjectiveMetricValue"
)["Candidates"]
index = 1
for candidate in candidates:
    print(
        str(index)
        + "  "
        + candidate["CandidateName"]
        + "  "
        + str(candidate["FinalAutoMLJobObjectiveMetric"]["Value"])
    )
    index += 1

ノートブック生成場所を確認

生成されるノートブックは次の2つです。

SageMakerAutopilotCandidateDefinitionNotebook.ipynb
- 前処理に使った設定(カテゴリカルデータのエンコーディング方法)や、ハイパーパラメータの調整時の設定を確認できます。
SageMakerAutopilotDataExplorationNotebook.ipynb
- 学習に使用されたデータセットの情報が記されています。中央値や平均値といった基本的な統計的情報や、データセットのそれぞれのカラムにおける型を確認できます。

SageMakerAutopilotCandidateDefinitionNotebook.ipynbのパス確認

sm.describe_auto_ml_job(AutoMLJobName=auto_ml_job_name)["AutoMLJobArtifacts"][
    "CandidateDefinitionNotebookLocation"
]

SageMakerAutopilotDataExplorationNotebook.ipynbのパス確認

sm.describe_auto_ml_job(AutoMLJobName=auto_ml_job_name)["AutoMLJobArtifacts"][
    "DataExplorationNotebookLocation"
]

終わりに

今回は、AutopilotをSageMaker Studioから実行してみました！

簡単に実行できるのでありがたいですね。また、ハイパーパラメータの調整もしてくれるので、予測精度を上げるための処理がなされているのもナイスですね。

今回はここまで。

それでは、また！

参考にしたサイト

【イベント開催のお知らせ】

2023/07/19(水)「DevelopersIO大阪」4年ぶりのオフライン開催決定！
AWSはもちろん、ChatGPTやデータ分析など幅広い分野の技術を取り扱った12のライブセッションをお送りします。
また、クラスメソッドのエンジニアと直接技術の話ができる相談ブースも開設いたします。
大阪オフィス所属のAWS Japan Top Engineersや元Ambassadorも大集合！

さらになんと、数量限定でクラスメソッド社員謹製の本格スパイスカレーを提供いたします！
入場無料、途中の入退室は自由です。ぜひご参加ください！

タイムスケジュール・お申込みはこちらから！ https://classmethod.connpass.com/event/286155/